package com.xxxx;

import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;

public class Hello01FlinkTime {
    public static void main(String[] args) throws Exception {

        //获取环境
        StreamExecutionEnvironment environment = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        //environment.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        environment.getConfig().setAutoWatermarkInterval(20000);
        //获取数据源
        DataStreamSource<String> source = environment.socketTextStream("localhost", 18880);

        source.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<String>() {

            //定义一个时间戳
            private long maxTimeStamp;
            //延迟时间
            private long maxLateTime = 5000;


            @Override
            public Watermark getCurrentWatermark() {
                return new Watermark(maxTimeStamp - maxLateTime);
            }

            @Override
            public long extractTimestamp(String line, long l) {
                return 0;
            }
        });
        //转换
        source.map(word -> Tuple2.of(word, 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(0)
                .timeWindow(Time.seconds(5))
                .sum(1)
                .print();
        //执行环境
        environment.execute("Hello01FlinkTime" + System.currentTimeMillis());

    }
}
